#include "cc_port.h"
#include "cc_shell.h"

cc_port ports[CC_CONFIG_CC_PORT_MAX_NUM];

cc_port_t cc_port_init(e_cc_port id, const char *name, size_t msg_num, size_t msg_size)
{
  cc_port_t port;
  // 如果当前端口已存在直接返回
  port = cc_port_get(id);
  if (port)
  {
    cc_log_d("port is exists\r\n");
    return port;
  }
  // 获取一个未使用的节点
  port = cc_port_get(CC_PORT_UNUSE);
  if (port == CC_NULL)
  {
    cc_log_e("port is full pealse add CC_CONFIG_CC_PORT_MAX_NUM\r\n");
    return CC_NULL;
  }
  if (id == CC_PORT_UNUSE)
  {
    cc_log_e("error port id\r\n");
    return CC_NULL;
  }
  memcpy(port->name, name, CC_CONFIG_PORT_NAME_MAX_LEN);
  port->state = CC_PORT_STATE_UNINIT;
  port->port_id = id;
  port->mem = cc_mempool_create(msg_num, msg_size);
  port->rx_que = cc_queue_create(msg_num, sizeof(cc_mnode));
  cc_list_init(&(port->sublist_head));
  cc_port_set_state(port, CC_PORT_STATE_INIT);
  return port;
}

cc_port_t cc_port_get(e_cc_port port_id)
{
  int i;

  for (i = 0; i < CC_CONFIG_CC_PORT_MAX_NUM; i++)
  {
    if (ports[i].port_id == port_id)
      return &ports[i];
  }
  return NULL;
}

cc_mnode_t cc_port_get_msg_node(cc_port_t port, uint32_t timeout)
{
  cc_mnode_t node = CC_NULL;
  if (port == CC_NULL)
  {
    cc_log_e("port is null\r\n");
    return CC_NULL;
  }
  if ((port->state == CC_PORT_STATE_UNINIT) && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port state not allow\r\n");
    return CC_NULL;
  }
  node = cc_mnode_get(port->mem, CC_NULL, port->data_size, timeout, CC_NULL);
  if (node == CC_NULL)
  {
    cc_log_e("node is null\r\n");
    return CC_NULL;
  }
  node->src_port_id = port->port_id;
  memset(node->mem, 0, port->data_size);
  return node;
}

// 点对点
int cc_port_send(cc_port_t port, e_cc_port target_id, cc_mnode_t node)
{
  if (node == CC_NULL)
  {
    return CC_ENOMEM;
  }
  if (port == CC_NULL)
  {
    if (node != CC_NULL)
    {
      cc_mnode_put(node);
    }
    cc_log_e("port is null\r\n");
    return CC_ENOMEM;
  }
  if (port->state != CC_PORT_STATE_CONN && port->port_id != CC_PORT_UNUSE)
  {
    if (node != CC_NULL)
    {
      cc_mnode_put(node);
    }
    cc_log_e("src port(%d) state(%d) not allow\r\n", port->port_id, port->state);
    return CC_ENOTCONN;
  }
  cc_port_t target_port;
  target_port = cc_port_get(target_id);
  if (target_port == CC_NULL)
  {
    if (node != CC_NULL)
    {
      cc_mnode_put(node);
    }
    cc_log_e("target port(%d) not exist\r\n", target_id);
    return CC_ENOMEM;
  }
  if (target_port->state != CC_PORT_STATE_CONN)
  {
    if (node != CC_NULL)
    {
      cc_mnode_put(node);
    }
    cc_log_e("target port(%d) state(%d) not allow\r\n", target_id, target_port->state);
    return CC_ENOTCONN;
  }
  node->src_port_id = (uint32_t)port->port_id;
  if (!cc_queue_send(target_port->rx_que, &node))
  {
    if (node != CC_NULL)
    {
      cc_mnode_put(node);
    }
    cc_log_e("send error\r\n");
    return CC_ETIMEDOUT;
  }

  return CC_EOK;
}

cc_mnode_t cc_port_recv_time(cc_port_t port, uint32_t timeout)
{
  if (port == CC_NULL)
  {
    cc_log_e("port is null\r\n");
    return CC_NULL;
  }
  if (port->state != CC_PORT_STATE_CONN && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port(%d) state(%d) not allow\r\n", port->port_id, port->state);
    return CC_NULL;
  }
  cc_mnode_t msg_node;
  if (!cc_queue_recv_time(port->rx_que, &msg_node, timeout))
  {
    cc_log_e("port(%d) recv failed", port->port_id);
    return CC_NULL;
  }
  return msg_node;
}

int cc_port_recv_listen_time(cc_port_t port, uint32_t timeout)
{
  if (port == CC_NULL)
  {
    cc_log_e("port is null\r\n");
    return CC_ENOMEM;
  }
  if (port->state != CC_PORT_STATE_CONN && port->port_id != CC_PORT_UNUSE && port->listener == CC_NULL)
  {
    cc_log_e("port(%d) state(%d) not allow\r\n", port->port_id, port->state);
    return CC_EBUSY;
  }
  cc_mnode_t msg_node;
  if (cc_queue_recv_time(port->rx_que, &msg_node, timeout))
  {
    port->listener((e_cc_port)msg_node->src_port_id, port->port_id, "recv_listen_mode", msg_node->mem);
    cc_mnode_put(msg_node);
    return CC_EOK;
  }
  else
  {
    return CC_ETIMEDOUT;
  }
}

int cc_port_recv_listen(cc_port_t port)
{
  return cc_port_recv_listen_time(port, CC_TIME_WAIT_FOREVER);
}

cc_mnode_t cc_port_recv(cc_port_t port)
{
  return cc_port_recv_time(port, CC_TIME_WAIT_FOREVER);
}
// 一对多 topic 为 publish 的话 就是广播
int cc_port_publish(cc_port_t port, const char *topic, cc_mnode_t node)
{
  if (port == CC_NULL && port->port_id != CC_PORT_UNUSE)
  {
    if (node != CC_NULL)
      cc_mnode_put(node);
    cc_log_e("port is null\r\n");
    return CC_ENOMEM;
  }
  if (strcmp(topic, "publish") == 0)
  {
    for (size_t i = 0; i < CC_CONFIG_CC_PORT_MAX_NUM; i++)
    {
      cc_port_t target_port = &ports[i];
      if (target_port->state == CC_PORT_STATE_CONN && target_port->listener != CC_NULL && target_port->port_id != port->port_id && target_port->port_id != CC_PORT_UNUSE)
      {
        target_port->listener((e_cc_port)node->src_port_id, target_port->port_id, "publish", node->mem);
      }
    }
  }
  else
  {
    for (size_t i = 0; i < CC_CONFIG_CC_PORT_MAX_NUM; i++)
    {
      cc_port_t target_port = &ports[i];
      if (!(target_port->state == CC_PORT_STATE_CONN && target_port->listener != CC_NULL && target_port->port_id != port->port_id && target_port->port_id != CC_PORT_UNUSE))
      {
        continue;
      }
      if (cc_list_isempty(&(target_port->sublist_head)))
      {
        continue;
      }
      void *data;
      cc_list_foreach(&(target_port->sublist_head))
      {
        cc_list_get_entry(&(target_port->sublist_head), data);
        struct cc_port_sub_list *temp = (struct cc_port_sub_list *)data;
        if (strcmp(temp->topic, topic) == 0)
        {
          target_port->listener((e_cc_port)node->src_port_id, target_port->port_id, topic, node->mem);
          break;
        }
      }
    }
  }
  if (node != CC_NULL)
    cc_mnode_put(node);
  return CC_EOK;
}

int cc_port_broadcast(cc_port_t port, cc_mnode_t node)
{
  if (port == CC_NULL && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port is null\r\n");
    return CC_ENOMEM;
  }
  return cc_port_publish(port, "publish", node);
}

int cc_port_unsubsribe(cc_port_t port, const char *topic)
{
  if (port == CC_NULL && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port is null\r\n");
    return CC_ENOMEM;
  }

  void *entry_data;
  cc_list_foreach(&(port->sublist_head))
  {
    if (cc_list_isempty(&(port->sublist_head)))
    {
      break;
    }
    cc_list_get_entry(&(port->sublist_head), entry_data);
    struct cc_port_sub_list *temp = (struct cc_port_sub_list *)entry_data;
    if (strcmp((char *)temp->topic, topic) == 0)
    {
      cc_list_remove(&temp->node);
      return CC_EOK;
    }
  }
  cc_log_e("don't have the topic,don't need to unsub\r\n");
  return CC_EINVAL;
}

int cc_port_subsribe(cc_port_t port, const char *topic)
{
  if (port == CC_NULL && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port is null\r\n");
    return CC_ENOMEM;
  }
  void *entry_data;
  cc_list_foreach(&(port->sublist_head))
  {
    if (cc_list_isempty(&(port->sublist_head)))
    {
      break;
    }
    cc_list_get_entry(&(port->sublist_head), entry_data);
    struct cc_port_sub_list *temp = (struct cc_port_sub_list *)entry_data;
    if (strcmp((char *)temp->topic, topic) == 0)
    {
      cc_log_e("do not repeat subscribe\r\n");
      return CC_EINVAL;
    }
  }

  struct cc_port_sub_list *sub_node = (struct cc_port_sub_list *)cc_malloc(sizeof(struct cc_port_sub_list));
  if (sub_node == CC_NULL)
  {
    cc_log_e("oom\r\n");
    return CC_ENOMEM;
  }
  if (strlen(topic) > CC_CONFIG_MSG_TOPIC_MAX_LEN)
  {
    cc_log_e("topic too long\r\n");
    free(sub_node);
    return CC_ENAVAIL;
  }

  cc_list_node_init(&sub_node->node);
  memcpy(sub_node->topic, topic, CC_CONFIG_MSG_TOPIC_MAX_LEN);
  cc_list_node_set_entry(&sub_node->node, sub_node);
  cc_list_insert_after(&port->sublist_head, &sub_node->node);
  return CC_EOK;
}

int cc_port_listen(cc_port_t port, cc_notify_callback_t cb)
{
  if (port == CC_NULL && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port is null\r\n");
    return CC_ENOMEM;
  }
  port->listener = cb;
  return CC_EOK;
}

// 设置连接状态
int cc_port_set_state(cc_port_t port, e_cc_port_state state)
{
  if (port == CC_NULL && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port is null\r\n");
    return CC_ENOMEM;
  }
  port->state = state;
  return CC_EOK;
}
e_cc_port_state cc_port_get_st(cc_port_t port)
{
  if (port == CC_NULL && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port is null\r\n");
    return CC_PORT_STATE_UNKNOWN;
  }
  return port->state;
}
// 设置调用者
int cc_port_set_user(cc_port_t port, void *user_data)
{
  if (port == CC_NULL && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port is null\r\n");
    return CC_PORT_STATE_UNKNOWN;
  }
  port->user_data = user_data;
  return CC_EOK;
}
void *cc_port_get_user(cc_port_t port)
{
  if (port == CC_NULL && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port is null\r\n");
    return CC_NULL;
  }
  return port->user_data;
}

int cc_set_send_end_cb(cc_mnode_t node, cc_port_send_end_t cb, void *user_data)
{
  if (node == CC_NULL)
  {
    cc_log_e("node is null\r\n");
    return CC_ENOMEM;
  }
  node->callback = (cc_mn_callback)cb;
  node->user_data = user_data;
  return CC_EOK;
}

int cc_port_disconnect(cc_port_t port)
{
  if (port == CC_NULL && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port is null\r\n");
    return CC_ENOMEM;
  }
  if (port->state != CC_PORT_STATE_DISCONN)
    port->state = CC_PORT_STATE_DISCONN;
  return CC_EOK;
}
int cc_port_deinit(cc_port_t port)
{
  if (port == CC_NULL && port->port_id != CC_PORT_UNUSE)
  {
    cc_log_e("port is null\r\n");
    return CC_ENOMEM;
  }
  port->state = CC_PORT_STATE_UNINIT;
  port->port_id = CC_PORT_UNUSE;

  cc_queue_destory(port->rx_que);
  cc_mempool_destroy(port->mem);
  port->data_size = 0;
  port->listener = 0;
  port->user_data = 0;
  cc_list_deinit(&(port->sublist_head));
  return CC_EOK;
}
static char result[1024] = ""; // 初始化结果数组
void show_port(char *param, int node)
{
  
  char temp[100];
  strcpy(result, "\r\n-----port info-----\r\n");
  for (int i = 0; i < CC_CONFIG_CC_PORT_MAX_NUM; i++)
  {
    if (ports[i].port_id != CC_PORT_UNUSE)
    {
      sprintf(temp, "port(%s) id(%d) state(%d)\r\n", ports[i].name, ports[i].port_id, ports[i].state);
      void *entry_data;
      strcat(result, temp);

      for (size_t list_index = 0; list_index < cc_list_get_length(&(ports[i].sublist_head)); list_index++)
      {
        if (cc_list_isempty(&(ports[i].sublist_head)))
        {
          break;
        }
        cc_list_get_entry(&(ports[i].sublist_head), entry_data);
        struct cc_port_sub_list *subnode = (struct cc_port_sub_list *)entry_data;
        sprintf(temp, " --sub_topic(%d):%s\r\n", list_index, subnode->topic);
        strcat(result, temp);
        memset(temp, 0, sizeof(temp));
      }
    }
  }
  strcat(result, "----------------------\r\n");
  // 回复
  cc_msg_resp(node, result, 0, 0);
}
CC_COMMAND_SHELL(show_port, "show now port");